package com.rrd.dw.azkaban;

import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.logging.Logger;

import com.rrd.dw.azkaban.flow.CommonJobProperties;
import com.rrd.dw.azkaban.flow.Edge;
import com.rrd.dw.azkaban.flow.Flow;
import com.rrd.dw.azkaban.flow.FlowProps;
import com.rrd.dw.azkaban.flow.Node;
import com.rrd.dw.azkaban.flow.Props;
import com.rrd.dw.azkaban.flow.SpecialJobTypes;

public class DirectoryFlowLoader {
    private static final DirFilter DIR_FILTER = new DirFilter();
    private static final String PROPERTY_SUFFIX = ".properties";
    private static final String JOB_SUFFIX = ".job";
    public static final String JOB_MAX_XMS = "job.max.Xms";
    public static final String MAX_XMS_DEFAULT = "1G";
    public static final String JOB_MAX_XMX = "job.max.Xmx";
    public static final String MAX_XMX_DEFAULT = "2G";

    private final Logger logger;
    private Props props;
    private HashSet<String> rootNodes;
    private HashMap<String, Flow> flowMap;
    private HashMap<String, Node> nodeMap;
    private HashMap<String, Map<String, Edge>> nodeDependencies;
    private HashMap<String, Props> jobPropsMap;

    // Flow dependencies for embedded flows.
    private HashMap<String, Set<String>> flowDependencies;

    private ArrayList<FlowProps> flowPropsList;
    private ArrayList<Props> propsList;
    private Set<String> errors;
    private Set<String> duplicateJobs;

    /**
     * Creates a new DirectoryFlowLoader.
     *
     * @param props
     *            Properties to add.
     * @param logger
     *            The Logger to use.
     */
    public DirectoryFlowLoader(Props props, Logger logger) {
        this.logger = logger;
        this.props = props;
    }

    /**
     * Returns the flow map constructed from the loaded flows.
     *
     * @return Map of flow name to Flow.
     */
    public Map<String, Flow> getFlowMap() {
        return this.flowMap;
    }

    /**
     * Returns errors caught when loading flows.
     *
     * @return Set of error strings.
     */
    public Set<String> getErrors() {
        return this.errors;
    }

    /**
     * Returns job properties.
     *
     * @return Map of job name to properties.
     */
    public Map<String, Props> getJobProps() {
        return this.jobPropsMap;
    }

    /**
     * Returns list of properties.
     *
     * @return List of Props.
     */
    public List<Props> getProps() {
        return this.propsList;
    }

    /**
     * Loads all flows from the directory into the project.
     *
     * @param project
     *            The project to load flows to.
     * @param baseDirectory
     *            The directory to load flows from.
     */
    public void loadProjectFlow(File baseDirectory) {
        this.propsList = new ArrayList<>();
        this.flowPropsList = new ArrayList<>();
        this.jobPropsMap = new HashMap<>();
        this.nodeMap = new HashMap<>();
        this.flowMap = new HashMap<>();
        this.errors = new HashSet<>();
        this.duplicateJobs = new HashSet<>();
        this.nodeDependencies = new HashMap<>();
        this.rootNodes = new HashSet<>();
        this.flowDependencies = new HashMap<>();

        // Load all the props files and create the Node objects
        this.loadProjectFromDir(baseDirectory.getPath(), baseDirectory, null);

//    jobPropertiesCheck(project);

        // Create edges and find missing dependencies
        this.resolveDependencies();

        // Create the flows.
        this.buildFlowsFromDependencies();

        // Resolve embedded flows
        this.resolveEmbeddedFlows();

    }

    private void loadProjectFromDir(String base, File dir, Props parent) {
        // Load all Job files. If there's a duplicate name, then we don't load
        File[] jobFiles = dir
            .listFiles(new SuffixFilter(DirectoryFlowLoader.JOB_SUFFIX));
        for (File file : jobFiles) {
            String jobName = this.getNameWithoutExtension(file);
            try {
                if (!this.duplicateJobs.contains(jobName)) {
                    if (this.jobPropsMap.containsKey(jobName)) {
                        this.errors.add(
                            "Duplicate job names found '" + jobName + "'.");
                        this.duplicateJobs.add(jobName);
                        this.jobPropsMap.remove(jobName);
                        this.nodeMap.remove(jobName);
                    } else {
                        Props prop = new Props(parent, file);
                        String relative = this.getRelativeFilePath(base,
                            file.getPath());
                        prop.setSource(relative);

                        Node node = new Node(jobName);
                        String type = prop.getString("type", null);
                        if (type == null) {
                            this.errors.add(
                                "Job doesn't have type set '" + jobName + "'.");
                        }

                        node.setType(type);

                        node.setJobSource(relative);
                        if (parent != null) {
                            node.setPropsSource(parent.getSource());
                        }

                        // Force root node
                        if (prop.getBoolean(CommonJobProperties.ROOT_NODE,
                            false)) {
                            this.rootNodes.add(jobName);
                        }

                        this.jobPropsMap.put(jobName, prop);
                        this.nodeMap.put(jobName, node);
                    }
                }
            } catch (IOException e) {
                this.errors.add("Error loading job file " + file.getName() + ":"
                    + e.getMessage());
            }
        }

        File[] subDirs = dir.listFiles(DirectoryFlowLoader.DIR_FILTER);
        for (File file : subDirs) {
            this.loadProjectFromDir(base, file, parent);
        }
    }

    private void resolveEmbeddedFlows() {
        for (String flowId : this.flowDependencies.keySet()) {
            HashSet<String> visited = new HashSet<>();
            this.resolveEmbeddedFlow(flowId, visited);
        }
    }

    private void resolveEmbeddedFlow(String flowId, Set<String> visited) {
        Set<String> embeddedFlow = this.flowDependencies.get(flowId);
        if (embeddedFlow == null) {
            return;
        }

        visited.add(flowId);
        for (String embeddedFlowId : embeddedFlow) {
            if (visited.contains(embeddedFlowId)) {
                this.errors.add("Embedded flow cycle found in " + flowId + "->"
                    + embeddedFlowId);
                return;
            } else if (!this.flowMap.containsKey(embeddedFlowId)) {
                this.errors.add("Flow " + flowId + " depends on "
                    + embeddedFlowId + " but can't be found.");
                return;
            } else {
                this.resolveEmbeddedFlow(embeddedFlowId, visited);
            }
        }

        visited.remove(flowId);
    }

    private void resolveDependencies() {
        // Add all the in edges and out edges. Catch bad dependencies and self
        // referrals. Also collect list of nodes who are parents.
        for (Node node : this.nodeMap.values()) {
            Props props = this.jobPropsMap.get(node.getId());

            if (props == null) {
//        logger.error("Job props not found!! For some reason.");
                continue;
            }

            List<String> dependencyList = props.getStringList(
                CommonJobProperties.DEPENDENCIES, (List<String>) null);

            if (dependencyList != null) {
                Map<String, Edge> dependencies = this.nodeDependencies
                    .get(node.getId());
                if (dependencies == null) {
                    dependencies = new HashMap<>();

                    for (String dependencyName : dependencyList) {
                        dependencyName = dependencyName == null ? null
                            : dependencyName.trim();
                        if (dependencyName == null
                            || dependencyName.isEmpty()) {
                            continue;
                        }

                        Edge edge = new Edge(dependencyName, node.getId());
                        Node dependencyNode = this.nodeMap.get(dependencyName);
                        if (dependencyNode == null) {
                            if (this.duplicateJobs.contains(dependencyName)) {
                                edge.setError(
                                    "Ambiguous Dependency. Duplicates found.");
                                dependencies.put(dependencyName, edge);
                                this.errors.add(
                                    node.getId() + " has ambiguous dependency "
                                        + dependencyName);
                            } else {
                                edge.setError("Dependency not found.");
                                dependencies.put(dependencyName, edge);
                                this.errors.add(
                                    node.getId() + " cannot find dependency "
                                        + dependencyName);
                            }
                        } else if (dependencyNode == node) {
                            // We have a self cycle
                            edge.setError("Self cycle found.");
                            dependencies.put(dependencyName, edge);
                            this.errors.add(node.getId() + " has a self cycle");
                        } else {
                            dependencies.put(dependencyName, edge);
                        }
                    }

                    if (!dependencies.isEmpty()) {
                        this.nodeDependencies.put(node.getId(), dependencies);
                    }
                }
            }
        }
    }

    private void buildFlowsFromDependencies() {
        // Find all root nodes by finding ones without dependents.
        HashSet<String> nonRootNodes = new HashSet<>();
        for (Map<String, Edge> edges : this.nodeDependencies.values()) {
            for (String sourceId : edges.keySet()) {
                nonRootNodes.add(sourceId);
            }
        }

        // Now create flows. Bad flows are marked invalid
        Set<String> visitedNodes = new HashSet<>();
        for (Node base : this.nodeMap.values()) {
            // Root nodes can be discovered when parsing jobs
            if (this.rootNodes.contains(base.getId())
                || !nonRootNodes.contains(base.getId())) {
                this.rootNodes.add(base.getId());
                Flow flow = new Flow(base.getId());
                Props jobProp = this.jobPropsMap.get(base.getId());

                // Dedup with sets
                @SuppressWarnings("unchecked")
                List<String> successEmailList = jobProp.getStringList(
                    CommonJobProperties.SUCCESS_EMAILS, Collections.EMPTY_LIST);
                Set<String> successEmail = new HashSet<>();
                for (String email : successEmailList) {
                    successEmail.add(email.toLowerCase());
                }

                @SuppressWarnings("unchecked")
                List<String> failureEmailList = jobProp.getStringList(
                    CommonJobProperties.FAILURE_EMAILS, Collections.EMPTY_LIST);
                Set<String> failureEmail = new HashSet<>();
                for (String email : failureEmailList) {
                    failureEmail.add(email.toLowerCase());
                }

                @SuppressWarnings("unchecked")
                List<String> notifyEmailList = jobProp.getStringList(
                    CommonJobProperties.NOTIFY_EMAILS, Collections.EMPTY_LIST);
                for (String email : notifyEmailList) {
                    email = email.toLowerCase();
                    successEmail.add(email);
                    failureEmail.add(email);
                }

                flow.addFailureEmails(failureEmail);
                flow.addSuccessEmails(successEmail);

                flow.addAllFlowProperties(this.flowPropsList);
                this.constructFlow(flow, base, visitedNodes);
                flow.initialize();
                this.flowMap.put(base.getId(), flow);
            }
        }
    }

    private void constructFlow(Flow flow, Node node, Set<String> visited) {
        visited.add(node.getId());

        flow.addNode(node);
        if (SpecialJobTypes.EMBEDDED_FLOW_TYPE.equals(node.getType())) {
            Props props = this.jobPropsMap.get(node.getId());
            String embeddedFlow = props.get(SpecialJobTypes.FLOW_NAME);

            Set<String> embeddedFlows = this.flowDependencies.get(flow.getId());
            if (embeddedFlows == null) {
                embeddedFlows = new HashSet<>();
                this.flowDependencies.put(flow.getId(), embeddedFlows);
            }

            node.setEmbeddedFlowId(embeddedFlow);
            embeddedFlows.add(embeddedFlow);
        }
        Map<String, Edge> dependencies = this.nodeDependencies
            .get(node.getId());

        if (dependencies != null) {
            for (Edge edge : dependencies.values()) {
                if (edge.hasError()) {
                    flow.addEdge(edge);
                } else if (visited.contains(edge.getSourceId())) {
                    // We have a cycle. We set it as an error edge
                    edge = new Edge(edge.getSourceId(), node.getId());
                    edge.setError("Cyclical dependencies found.");
                    this.errors
                        .add("Cyclical dependency found at " + edge.getId());
                    flow.addEdge(edge);
                } else {
                    // This should not be null
                    flow.addEdge(edge);
                    Node sourceNode = this.nodeMap.get(edge.getSourceId());
                    this.constructFlow(flow, sourceNode, visited);
                }
            }
        }

        visited.remove(node.getId());
    }

    private String getNameWithoutExtension(File file) {
        String filename = file.getName();
        int index = filename.lastIndexOf('.');

        return index < 0 ? filename : filename.substring(0, index);
    }

    private String getRelativeFilePath(String basePath, String filePath) {
        return filePath.substring(basePath.length() + 1);
    }

    private static class DirFilter implements FileFilter {
        @Override
        public boolean accept(File pathname) {
            return pathname.isDirectory();
        }
    }

    private static class SuffixFilter implements FileFilter {
        private String suffix;

        public SuffixFilter(String suffix) {
            this.suffix = suffix;
        }

        @Override
        public boolean accept(File pathname) {
            String name = pathname.getName();

            return pathname.isFile() && !pathname.isHidden()
                && name.length() > this.suffix.length()
                && name.endsWith(this.suffix);
        }
    }

}
